package cn.zzdt4j.config.springboot.starter.refresher;

import cn.hutool.core.util.StrUtil;
import cn.zzdt4j.common.toolkit.JSONUtil;
import cn.zzdt4j.threadpool.dynamic.model.config.properties.BootstrapConfigProperties;
import io.etcd.jetcd.*;
import io.etcd.jetcd.kv.GetResponse;
import io.etcd.jetcd.watch.WatchEvent;
import io.etcd.jetcd.watch.WatchResponse;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ExecutionException;

/**
 * Etcd refresh handler
 *
 * @author by <a href="mailto:ligang941012@gmail.com">gang.Li</a>
 * @since 2024/5/13 14:36
 */
@Slf4j
public class EtcdRefresherHandler extends AbstractConfigThreadPoolDynamicRefresh {

    private Client client;

    private static final String PASSWORD = "password";

    private static final String ENDPOINTS = "endpoints";

    private static final String AUTHORITY = "authority";

    private static final String USER = "user";

    private static final String CHARSET = "charset";

    private static final String KEY = "key";

    @SneakyThrows(value = {InterruptedException.class, ExecutionException.class})
    @Override
    public void registerListener() {
        BootstrapConfigProperties actualBootstrapConfigProperties = (BootstrapConfigProperties) bootstrapConfigProperties;
        final Map<String, String> etcd = actualBootstrapConfigProperties.getEtcd();
        final String key = etcd.get(KEY);
        final Charset charset = StrUtil.isBlank(etcd.get(CHARSET)) ? StandardCharsets.UTF_8 : Charset.forName(etcd.get(CHARSET));
        // init etcd client.
        initClient(etcd, charset);
        // TODO Currently only supports json
        final GetResponse getResponse = client.getKVClient().get(ByteSequence.from(key, charset)).get();
        final KeyValue keyValue = getResponse.getKvs().get(0);
        if (Objects.isNull(keyValue)) return;
        client.getWatchClient().watch(ByteSequence.from(key, charset), new Watch.Listener() {
            @Override
            public void onNext(WatchResponse watchResponse) {
                final WatchEvent watchEvent = watchResponse.getEvents().get(0);
                final WatchEvent.EventType eventType = watchEvent.getEventType();
                // todo Currently only supports json
                if (Objects.equals(eventType, WatchEvent.EventType.PUT)) {
                    final KeyValue keyValue1 = watchEvent.getKeyValue();
                    final String value = keyValue1.getValue().toString(charset);
                    final Map map = JSONUtil.parseObject(value, Map.class);
                    dynamicRefresh(keyValue1.getKey().toString(charset), map);
                }
            }

            @Override
            public void onError(Throwable throwable) {
                log.error("Dynamic thread pool etcd config watcher exception ", throwable);
            }

            @Override
            public void onCompleted() {
                log.info("Dynamic thread pool etcd config key refreshed, config ket {} ", key);
            }
        });

    }

    private void initClient(Map<String, String> etcd, Charset charset) {
        // TODO
        if (Objects.isNull(client)) {
            final String user = etcd.get(USER);
            final String password = etcd.get(PASSWORD);
            final String authority = etcd.get(AUTHORITY);
            final String endpoints = etcd.get(ENDPOINTS);
            ClientBuilder clientBuilder = Client.builder().endpoints(etcd.get(","));
            client = StrUtil.isAllNotBlank(user, password) ? clientBuilder.user(ByteSequence.from(user, charset))
                    .password(ByteSequence.from(password, charset))
                    .authority(authority)
                    .build() : clientBuilder.build();
        }
    }
}
